#include "sssp.h"

/* this function minics PEval of sssp
 * Input:
 *      IN_messages: igraph_vector_ptr_t  of struct my_pair, 
 *      namely, (u, dist[u]) .
 *      
 *      OUT_messages: igraph_vector_ptr_t  of igraph_vector_ptr_t *
 *      the reason not choosing igraph_matrix_t is that, it is in 
 *      fact igraph_vector_t, so individual rows cannot be extracted.
 *      
 *      Partial_result: igraph_vector_int_t of len innerv_size of my_pair
 *      the reason is to provide hashmap-like directed access to
 *      vertex of local_vid.
 *      
 *      TODO 
 *      Fragment: this will be a struct that provides 
 *          o: can tell whether a vertex isBorderVertexWithOutgoingEdge
 *          o: can tell whether a vertex isInnerVertex
 *          o: can getOutgoingEdge of u
 *          o: can getInnerVertex
 *          o: can get all the Fragment it communicates to based on
 *             a isBorderVertexWithOutgoingEdge u.
 *          
 *         although hashmap is capable of these operations, we use  
 *         vector instead as the global graph has been localized
 *         in a fragment 
 */

/* sssp_results_init:
 * construct a igraph_vector_int_t on 
 * args[i].result passed by grape's init 
 */

char *sssp_doc = "sssp -- Single Shortest Path Agorithm using GRAPE MODEL";           

char *sssp_args_doc = "";

struct sssp_res_t *sssp_make_res(int para, int dist)
{
        struct sssp_res_t *pr=xmalloc(sizeof(struct sssp_res_t));

        pr->dist=dist;
        pr->para=para;
        return pr;
}

int sssp_results_init(struct args_t *args, void **result)
{
        *result=xmalloc(sizeof(igraph_vector_ptr_t));
        igraph_vector_ptr_init(*result, args->fragment->innerv_size);
        // transfer ownership to sssp_assembled_result later
        return 0;
}

/* before develop sssp_peval, first check whether running dijktra
 * on the whole graph yield the same result as igraph_simple_paths
 */
void sssp_serial(my_adjlist_t *adjlist,int source,igraph_vector_int_t *result)
{
        igraph_indheap_t heap;
        igraph_vector_bool_t visited;
        igraph_vector_ptr_t *al;
        int dist,src,dst,attr;
        struct vertex_t *pv;
        struct edge_t *pe;
        
        // zero based
        source--;

        int vn= adjlist->vn;
        igraph_vector_int_init(result,vn);
        igraph_vector_bool_init(&visited,vn);
        igraph_indheap_init(&heap,vn);

        // deal with source
        for(int i=0;i<vn;++i)
                *igraph_vector_int_e_ptr(result, i)=INT_MAX;
        VECTOR(*result)[source]=0;
        igraph_indheap_push_with_index(&heap,source,0);
        
        // main loop
        while(!igraph_indheap_empty(&heap))
        {
                int u=igraph_indheap_max_index(&heap);
                dist= - igraph_indheap_delete_max(&heap);

                if(VECTOR(visited)[u])
                        continue;
                VECTOR(visited)[u]=1;

                al=my_adjlist_get(adjlist,u);
                int size=igraph_vector_ptr_size(al);

                for(int i=0;i<size;++i) {
                        pe=igraph_vector_ptr_e(al,i);
                        dst=pe->dst;
                        attr=pe->attr;
                        if(attr + dist < igraph_vector_int_e(result,dst))
                        {
                                *igraph_vector_int_e_ptr(result,dst) = attr + dist;
                                igraph_indheap_push_with_index(&heap,dst,(igraph_real_t) -(attr + dist));
                        }
                }
        }

        igraph_indheap_destroy(&heap);
        igraph_vector_bool_destroy(&visited);
}

int sssp_load_query(FILE *q, igraph_vector_ptr_t *qs)
{

        // to be simple, source vertices are list per line
        int vn=grape_get_vn();
        void *source ;
        while(1== fscanf(q,"%d",&source))
        {
                if((int) source <= 0 || (int) source > vn) {
                        return 1;
                }
                igraph_vector_ptr_push_back(qs,  source);
        }
        
        for(int i=0;i<igraph_vector_ptr_size(qs);++i) {
                zlog_debug(grape_inter_logger, "%d\n", (int) igraph_vector_ptr_e(qs,i));
        }
        return 0;
}

void *sssp_copy_query(void *q)
{
       return q;
}


struct sssp_msg_t *sssp_make_msg (int vid, int dist, int para)
{
        struct sssp_msg_t*pp=xmalloc(sizeof(struct sssp_msg_t));
        pp->dist=dist;
        pp->vid=vid;
        pp->para=para;
        return pp;
}


void sssp_print_msg(FILE *s, void *ptr)
{
        struct sssp_msg_t*p=ptr;
        fprintf(s,"sssp_msg_t(vid=%d,dist=%d,para=%d)\n", 
                        p->vid,
                        p->dist,
                        p->para);
}

/* sssp_peval:
 * do sssp partial evaluation on a fragment 
 * Input: RESULT, FRAGMENT, MSG, GADJLIST
 * Output: NONE
 */

void *sssp_peval(void *ptr)
{

        // get actuall argumets from void*
        struct args_t *args=ptr;
        struct Fragment *fragment=grape_get_fragment(args);
        igraph_vector_ptr_t *result=grape_get_partial_result(args);
        struct msg_buf_t *msg=grape_get_messages_buffer(args);
        int src=(int)grape_get_query(args);
        src-=1;

        struct vertex_t *source=igraph_vector_ptr_e(args->vertices,src);
        int myfid=grape_get_fid(args);
        my_adjlist_t *adjlist=fragment->global_adjlist;
        int vn=grape_get_vn();
        int innerv_size=fragment_get_innerv_size(fragment);
        
        struct sssp_msg_t *pp;
        struct vertex_t *pv;
        struct edge_t *pe;
        struct sssp_res_t *pr;

 
        for(int i=0;i<innerv_size;++i) {
                pr=sssp_make_res(SSSP_NOPARA, INT_MAX);
                igraph_vector_ptr_set(result,i, pr);
        }
        if(source->fid != myfid) {
                if(grape_data.parallel)
                        pthread_exit(NULL);
                else 
                        return 0;
        }

        igraph_indheap_t heap;
        igraph_vector_bool_t visited;
        igraph_vector_bool_init(&visited, innerv_size);
        igraph_indheap_init(&heap,0);
        igraph_indheap_push_with_index(&heap, source->locid, 0);
        pr=igraph_vector_ptr_e(result, source->locid);
        pr->dist=0;

        // main loop 
        while(!igraph_indheap_empty(&heap))
        {
                int u=igraph_indheap_max_index(&heap);
                int distu= (int)- igraph_indheap_delete_max(&heap);
                pr=igraph_vector_ptr_e(result, u);
                pv=igraph_vector_ptr_e(fragment->innerv,u);
                int vidu=pv->vid;


                if(VECTOR(visited)[u]) 
                        continue;

                VECTOR(visited)[u]=1;

                // issue OUT_messages
                struct vertex_t *pv=fragment_get_innerv_at(fragment,u);
                int nfid=vertex_get_comm_size(pv);

                for(int i=0;i<nfid;++i) {
                        int fid=vertex_get_comm_at(pv, i);
                        pp=sssp_make_msg(pv->vid,distu,pr->para);
                        add_out_msg_fid(msg, fid, pp);
                }
                
                // visit the local adjlist of u
                igraph_vector_ptr_t *local_adjlist=my_adjlist_get(fragment->local_adjlist,u);
                int adjsize=igraph_vector_ptr_size(local_adjlist);

                for(int i=0;i<adjsize;++i) {
                        pe=igraph_vector_ptr_e(local_adjlist,i);
                        int dst=pe->dst;
                        int attr=pe->attr;
                        pr=igraph_vector_ptr_e(result,dst);
                        int distv=pr->dist;

                        if(distv > distu + attr) {
                                int new_dist=distu+attr;
                                pr->dist=new_dist;
                                pr->para=vidu;
                                igraph_indheap_push_with_index(&heap,dst,(igraph_real_t) -  new_dist);
                        }
                }
        }

        igraph_indheap_destroy(&heap);
        igraph_vector_bool_destroy(&visited);

        if(grape_data.parallel)
                pthread_exit(NULL);
        else 
                return 0;
}


/* sssp_inceval:
 * updates the partial_result of fragment i
 * based on IN msg designated to i and generates
 * OUT msg to the fragments it communicates to
 * Input: RESULT, FRAGMENT, MSG, GADJLIST
 * Output: NONE
 */

void *sssp_inceval(void *ptr)
{

        /* get needed argumets */ 
        struct args_t *args=ptr;
        struct msg_buf_t *msg=grape_get_messages_buffer(args);
        igraph_vector_ptr_t *result=grape_get_partial_result(args);
        struct Fragment *fragment=grape_get_fragment(args);
        int myfid=grape_get_fid(args);
        my_adjlist_t *global_adjlist=fragment->global_adjlist;


        struct sssp_msg_t *pp;
        struct vertex_t *pv;
        struct edge_t *pe;
        struct sssp_res_t *pr;

        int inmsg_size=get_in_msg_size(msg);
        igraph_indheap_t heap;
        igraph_indheap_init(&heap, 0);
        igraph_vector_bool_t visited;
        igraph_vector_bool_init(&visited,fragment->innerv_size);

        /* updates partial_result of myfid */
        for(int i=0;i<inmsg_size;++i) {
                pp=get_in_msg_at(msg, i);
                igraph_vector_ptr_t *adjlist=my_adjlist_get(global_adjlist, pp->vid);
                int adjlist_size=igraph_vector_ptr_size(adjlist);

                /* visits the global_adjlist of i to find all the end points that falls into myfid */
                for(int j=0;j<adjlist_size;++j) {
                        pe=igraph_vector_ptr_e(adjlist,j);
                        pv=igraph_vector_ptr_e(args->vertices,pe->dst);
                        if(pv->fid != myfid)
                                continue;

                        /* Converts global_vid to local_vid */
                        int local_vid=pv->locid;
                        pr=igraph_vector_ptr_e(result,local_vid);
                        int dist=pr->dist;
                        if(pe->attr + pp->dist < dist) {
                                int new_dist=pe->attr + pp->dist;
                                pr->para=pp->vid;
                                pr->dist=new_dist;
                                igraph_indheap_push_with_index(&heap, local_vid, (igraph_real_t) - new_dist);
                        }
                }
        }
        

        /* main loop */
        while(!igraph_indheap_empty(&heap))
        {
                int u=igraph_indheap_max_index(&heap);
                int distu= (int) - igraph_indheap_delete_max(&heap);
                pr = igraph_vector_ptr_e(result, u);
                pv=igraph_vector_ptr_e(fragment->innerv,u);
                int vidu=pv->vid;
                

                if(VECTOR(visited)[u])
                        continue;
                VECTOR(visited)[u]=1;

                assert(pv->locid == u); 
                assert(pr->dist == distu);

                // issue OUT_messages

                pv=fragment_get_innerv_at(fragment,u);
                int nfid=vertex_get_comm_size(pv);
                
                // visits BorderVertexWithOutgoingEdge innerv 
                for(int i=0;i<nfid;++i) {
                        int fid=vertex_get_comm_at(pv,i);

                        /* Converts to global_vid */
                        pp=sssp_make_msg(pv->vid, distu, pr->para);
                        add_out_msg_fid(msg, fid, pp);
                }
                
                // visit the local adjlist of u
                igraph_vector_ptr_t *adjlistu=my_adjlist_get(fragment->local_adjlist,u);
                int adjsize=igraph_vector_ptr_size(adjlistu);

                for(int i=0;i<adjsize;++i) {
                        pe=igraph_vector_ptr_e(adjlistu,i);
                        int dst=pe->dst;
                        int attr=pe->attr;
                        pr =igraph_vector_ptr_e(result,dst);
                        int distv=pr->dist;

                        if(distv > distu + attr) {
                                int new_dist=distu + attr;

                                assert (pr->dist > new_dist);

                                pr->dist=new_dist;
                                pr->para=vidu;
                                igraph_indheap_push_with_index(&heap,dst,(igraph_real_t) -new_dist);
                        }
                }
        }
        igraph_vector_bool_destroy(&visited);
        igraph_indheap_destroy(&heap);

        if(grape_data.parallel)
                pthread_exit(NULL);
        else 
                return 0;
}


/* sssp_assemble:
 * assemble all the partial_results and writes it to file
 * Input : argumets array
 * Ouput: assembled_result in the format of
 * vn line, each line -> vid    dist
 * Time: O(vn)
 */

igraph_vector_ptr_t  sssp_assembled_result;
void* sssp_assemble(struct args_t *args, int nfrag)
{
        int vn=grape_get_vn();
        igraph_vector_ptr_t *partial_result;
        FILE *output_file=grape_get_output_file();
        struct sssp_res_t *pr;

        igraph_vector_ptr_init(&sssp_assembled_result,vn);
        for(int i=0;i<nfrag;++i) {
                partial_result=args[i].result ;
                int size=igraph_vector_ptr_size(partial_result);
                struct Fragment *fragment=args[i].fragment;
                for(int j=0;j<size;++j) {
                        int local_vid=j;
                        struct vertex_t *pv=igraph_vector_ptr_e(fragment->innerv,j);
                        int global_vid=pv->vid;
                        pr =igraph_vector_ptr_e(partial_result,j);
                        VECTOR(sssp_assembled_result)[global_vid]=pr;
                }
        }
        
        
        for(int i=0;i<vn;++i) {
                pr=VECTOR(sssp_assembled_result)[i];
                fprintf(output_file, "%d\t%d\n", i+1, pr->dist);
        }
        igraph_vector_ptr_destroy(&sssp_assembled_result);
        return NULL;

}

void sssp_results_destroy(void *p)
{
        igraph_vector_ptr_destroy(p);
        free(p);
}

struct grape_plugin_t sssp_plugin = 
{
        sssp_results_init,
        sssp_results_destroy,
        printl_pair,
        print_pair,
        free,
        sssp_load_query,
        free,
        sssp_copy_query,

        sssp_peval,
        sssp_inceval,
        sssp_assemble
};

int verify_sssp(FILE *dimacs, int source, igraph_matrix_t *result)
{
        igraph_t undirected_graph;
        zlog_category_t *mylog = zlog_get_category("mylog");
        int vn, rs;

        rs = igraph_read_graph_dimacs(&undirected_graph,
                        dimacs,
                        0,0,0,0,0,IGRAPH_UNDIRECTED);
        if(rs != IGRAPH_SUCCESS) {
                zlog_error(mylog,"read dimacs failed");
                return -1;
        }
        vn = igraph_vcount(&undirected_graph);
        igraph_vs_t from,to;
        igraph_vs_1(&from, source);
        igraph_vs_all(&to);
        igraph_matrix_init(result,1,vn);

        rs = igraph_shortest_paths(&undirected_graph,result,from,to,IGRAPH_ALL);

        if(rs != IGRAPH_SUCCESS) {
                zlog_error(mylog,"igraph_shortest_paths failed");
                return -1;
        }
        // cleanup
        igraph_destroy(&undirected_graph);

        return 0;
}

